package com.hu.api

import org.apache.flink.api.common.accumulators.IntCounter
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @Author: hujianjun
 * @Create Date: 2021/1/19 11:20
 * @Describe: 无界流中获取累加结果值可以通过转换操作返回累计器数值，比如map,flatmap的输出中包含累加值
 */
object IntCounterUnbounded {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()

    val inputStream = env.socketTextStream("localhost", 6666).uid("s1").name("读取socket数据源")

    inputStream.flatMap(new RichFlatMapFunction[String, IntCounter] {
      private var inc: IntCounter = null

      override def open(parameters: Configuration): Unit = {
        inc = new IntCounter();
        getRuntimeContext().addAccumulator("word_nums", this.inc);
      }

      override def flatMap(value: String, out: Collector[IntCounter]): Unit = {
        val strings = value.split(" ")
        strings.foreach(_ => {
          this.inc.add(1)
        })

        //无界流需要将累加结果发送出去
        out.collect(this.inc)
      }
    }).name("空格拆分输入内容，统计出现的总词数").print()

    //获取逻辑执行图的json https://flink.apache.org/visualizer
    env.getExecutionPlan
    println(env.execute("实现无界流累加器demo").getAccumulatorResult("word_nums"))
  }
}
